package main

import (
	"fmt"
	"github.com/Shopify/sarama"
	"log"
	"note/kafka_demo/broker"
	"note/kafka_demo/broker/kafka"
	"sync"
)

func main() {
	config := kafka.DefaultClusterConfig
	config.Version = sarama.V0_10_2_1
	client := kafka.NewBroker(kafka.ClusterConfig(config))

	err := client.Connect()
	if err != nil {
		return
	}
	var wg sync.WaitGroup
	wg.Add(1)
	client.Subscribe("my-kafka", func(event broker.Event) error {
		fmt.Println(event.Message())
		_ = event.Ack()
		wg.Done()
		return nil
	}, func(options *broker.SubscribeOptions) {
		options.Queue = "example"
	})
	wg.Wait()
}

func test2() {
	//config := sarama.NewConfig()
	//config.Version = sarama.V0_10_2_1
	//config.Consumer.Return.Errors = true
	//config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRoundRobin
	//config.Consumer.Offsets.Initial = sarama.OffsetOldest
	//
	//cs, err := sarama.NewClient([]string{"127.0.0.1:9092"}, config)
	//
	///**
	// * Setup a new Sarama consumer group
	// */
	//consumer := Consumer{
	//	ready: make(chan bool),
	//}
	//
	//ctx, cancel := context.WithCancel(context.Background())
	//client, err := sarama.NewConsumerGroupFromClient("example", cs)
	//if err != nil {
	//	log.Panicf("Error creating consumer group client: %v", err)
	//}
	//
	//wg := &sync.WaitGroup{}
	//wg.Add(1)
	//go func() {
	//	defer wg.Done()
	//	for {
	//		if err := client.Consume(ctx, []string{"my-kafka"}, &consumer); err != nil {
	//			log.Panicf("Error from consumer: %v", err)
	//		}
	//		// check if context was cancelled, signaling that the consumer should stop
	//		if ctx.Err() != nil {
	//			return
	//		}
	//		consumer.ready = make(chan bool)
	//	}
	//}()
	//
	//<-consumer.ready // Await till the consumer has been set up
	//log.Println("Sarama consumer up and running!...")
	//
	//sigterm := make(chan os.Signal, 1)
	//signal.Notify(sigterm, syscall.SIGINT, syscall.SIGTERM)
	//select {
	//case <-ctx.Done():
	//	log.Println("terminating: context cancelled")
	//case <-sigterm:
	//	log.Println("terminating: via signal")
	//}
	//cancel()
	//wg.Wait()
	//if err = client.Close(); err != nil {
	//	log.Panicf("Error closing client: %v", err)
	//}
}

func native() {
	//	fmt.Printf("consumer_test")
	//
	//	config := sarama.NewConfig()
	//	config.Consumer.Return.Errors = true
	//	config.Version = sarama.V0_9_0_0
	//
	//	// consumer
	//	consumer, err := sarama.NewConsumer([]string{"127.0.0.1:9092"}, config)
	//	if err != nil {
	//		fmt.Printf("consumer_test create consumer error %s\n", err.Error())
	//		return
	//	}
	//	defer consumer.Close()
	//
	//	partition_consumer, err := consumer.ConsumePartition("my-kafka", 0, sarama.OffsetOldest)
	//	if err != nil {
	//		fmt.Printf("try create partition_consumer error %s\n", err.Error())
	//		return
	//	}
	//	defer partition_consumer.Close()
	//
	//	for {
	//		select {
	//		case msg := <-partition_consumer.Messages():
	//			fmt.Printf("msg offset: %d, partition: %d, timestamp: %s, value: %s\n",
	//				msg.Offset, msg.Partition, msg.Timestamp.String(), string(msg.Value))
	//
	//		case err := <-partition_consumer.Errors():
	//			fmt.Printf("err :%s\n", err.Error())
	//		}
	//	}
}

// Consumer represents a Sarama consumer group consumer
type Consumer struct {
	ready chan bool
}

// Setup is run at the beginning of a new session, before ConsumeClaim
func (consumer *Consumer) Setup(sarama.ConsumerGroupSession) error {
	// Mark the consumer as ready
	close(consumer.ready)
	return nil
}

// Cleanup is run at the end of a session, once all ConsumeClaim goroutines have exited
func (consumer *Consumer) Cleanup(sarama.ConsumerGroupSession) error {
	return nil
}

// ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages().
func (consumer *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {

	// NOTE:
	// Do not move the code below to a goroutine.
	// The `ConsumeClaim` itself is called within a goroutine, see:
	// https://github.com/Shopify/sarama/blob/master/consumer_group.go#L27-L29
	for message := range claim.Messages() {
		log.Printf("Message claimed: value = %s, timestamp = %v, topic = %s", string(message.Value), message.Timestamp, message.Topic)
		session.MarkMessage(message, "")
	}

	return nil
}
